package net.bwie.flink.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * HBase数据库工具类，实现数据插入insert、查询获取get、创建表（判断是否存在）
 * @author xuanyu
 * @date 2025/10/31
 */
public class HbaseUtil {

	private static Connection connection = null ;
	private Table table = null ;

	static {
		try{
			// 1.创建配置对象，设置属性值，比如连接ZK集群信息
			org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create() ;
			conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
			conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
			// 2.实例化Connection连接
			connection = ConnectionFactory.createConnection(conf) ;
		}catch (Exception e){
			e.printStackTrace();
		}
	}

	/**
	 * 将数据插入HBase表，指定相关字段名称，其中数据为json字符串
	 * @param tableName 表名称
	 * @param primaryName 主键字段名称
	 * @param familyName 列簇名称
	 * @param columnNames 列名称，多个列名称之间使用逗号隔开
	 * @param jsonValue 列值，json字符串
	 */
	public static void putData(String tableName,
	                           String primaryName,
	                           String familyName,
	                           String columnNames, String jsonValue) throws Exception{
		// 3. 实例化Table对象
		Table table = connection.getTable(TableName.valueOf(tableName));
		// 4. 添加数据
		JSONObject jsonObject = JSON.parseObject(jsonValue);
		// put对象
		String rowKey = jsonObject.get(primaryName).toString();
		Put put = new Put(Bytes.toBytes(rowKey));
		// 添加列
		for(String columnName: columnNames.split(",")){
			// 依据列名称获取列值
			Object columnObject = jsonObject.get(columnName);
			// 当列值不为null时，插入hbase表
			if(null != columnObject){
				put.addColumn(
					Bytes.toBytes(familyName), Bytes.toBytes(columnName) , Bytes.toBytes(columnObject.toString())
				);
			}
		}
		// 5. 插入
		table.put(put);
		table.close();
	}

	/**
	 * 依据表名称和列簇名称，在hbase数据库中创建表，先判断表是否存在
	 * @param tableName 表名称
	 * @param familyName 列簇名称
	 */
	public static void createHbaseTable(String tableName, String familyName) throws Exception {
		TableName table = TableName.valueOf(tableName);
		// 1. 管理对象
		Admin admin = connection.getAdmin();
		// 2. 表不存在时，创建表
		if(!admin.tableExists(table)){
			// 表描述符-构造器
			TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(table);
			// 列簇描述符-构造器
			ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder
				.newBuilder(Bytes.toBytes(familyName));
			// 表添加列簇
			tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
			// 创建表
			admin.createTable(tableDescriptorBuilder.build());
		}
		// 3. 关闭
		admin.close();
	}

	/**
	 * 依据RowKey获取HBase表中数据
	 * @param tableName 表名称
	 * @param rowKey 主键
	 * @param familyName 列簇名称
	 * @param columnName 列名称
	 */
	public static String getData(String tableName, String rowKey,
	                             String familyName, String columnName) throws Exception{
		// 3.实例化Table对象
		Table table = connection.getTable(TableName.valueOf(tableName));
		// 4. 获取数据
		Get get = new Get(Bytes.toBytes(rowKey));
		Result result = table.get(get);
		// 5. 获取字段之
		byte[] resultValue = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
		table.close();
		// 6. 返回
		return Bytes.toString(resultValue) ;
	}


	/**
	 * 关闭链接
	 */
	public static void close() throws Exception {
		if(null != connection){
			connection.close();
		}
	}

}
